package com.moli.iov.handle.worker.his.util;

import com.moli.iov.handle.worker.his.hbase.IHbaseService;
import com.moli.iov.handle.worker.his.hbase.dto.create.InsertBatchCreateDto;
import com.moli.iov.handle.worker.his.lmax.Disruptor;
import com.moli.iov.handle.worker.his.lmax.DisruptorTask;
import com.moli.iov.handle.worker.his.mongodb.Mongo;
import com.moli.iov.protocol.codec.DataDecoder;
import com.moli.iov.protocol.codec.ReqStreamData;
import com.moli.iov.protocol.enums.CommandTag;
import com.moli.iov.protocol.enums.DateTimeTag;
import com.moli.iov.protocol.model.realtime.*;
import com.moli.iov.util.ByteArrayUtil;
import com.moli.iov.util.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 *
 * Kafka consumer group listeners common tools.
 *
 * @author OAK
 * @since 2020/01/16 16:44 PM.
 * @version 3.0.1
 *
 */
@Slf4j
@Component
public class ConsumerCommonServer implements InitializingBean {

    @Value("${hbase.table.family}")
    private  String HBASE_FAMILY;

    @Resource
    private IHbaseService iHbaseService;
    @Resource
    private Mongo mongoService;

    @Resource(name = "realtimeDisruptorService")
    private Disruptor<ConcurrentMap, DisruptorTask> disruptor;

    @Resource(name = "realtimeHandler")
    private RealtimeHandler realtimeHandler;

    /**
     * 生成hbase的rowkey
     * @param vin
     * @param date
     * @return
     */
    public  String getRowkey(String vin,Date date){
        //hbase行键
        String rowKey=vin+"_"+(Long.MAX_VALUE-date.getTime());
        return rowKey;
    }


    /**
     * 插入hbase数据
     * @param tableName
     * @param map
     * @throws Exception
     */
    public  void  insertBatchData(String tableName,HashMap<String,HashMap<String,Object>> map) throws Exception {
        InsertBatchCreateDto insertBatchCreateDto=new InsertBatchCreateDto();
        insertBatchCreateDto.setFamily(HBASE_FAMILY);
        insertBatchCreateDto.setTableName(tableName);
        insertBatchCreateDto.setRowKeyAndColumnAndValue(map);
        iHbaseService.insertBatchData(insertBatchCreateDto);
    }

    /**
     * 部分报文相同简单的处理逻辑
     * @param list
     * @param acknowledgment
     * @param hbaseTable
     */
    public void simpleCommonProcess(List<ConsumerRecord<String, Object>> list, Acknowledgment acknowledgment,String hbaseTable){
        //存储容器 <rowkey,<colum,value>>
        HashMap<String, HashMap<String, Object>> topicMap = new HashMap<>(0);
        String topic="";
        for (ConsumerRecord record : list) {
            topic = record.topic();
            ReqStreamData reqStreamData = (ReqStreamData) record.value();
            if (null != reqStreamData) {
                String rowKey = getRowkey(record.key().toString(), reqStreamData.getCollectDate());
                HashMap<String, Object> map = new HashMap<String, Object>(0);
                map.put("messageContent", reqStreamData.dataBytes);
                topicMap.put(rowKey, map);
            } else {
                log.error("[车辆topic={}主题报文为空]", topic);
            }
        }
        //保存入hbase中
        try {
            if (!topicMap.isEmpty()) {
                insertBatchData(hbaseTable, topicMap);
                acknowledgment.acknowledge();
                //log.info("{}---------------成功完成消费{}条数据",topic,list.size());
            }
        } catch (Exception e) {
            log.error("hbase插入保存{}数据共{}条失败.....{}",topic,list.size(), e);
        }
    }

    /**
     * Real time message handle event for kafka consume group listeners.
     * @param list Consume record collection.
     * @param acknowledgment  Acknowledgment to commit offset for real time message.
     * @param tableName HBase real time message table name.
     */
    public void realtimeHandle(List<ConsumerRecord<String, Object>> list, Acknowledgment acknowledgment,String tableName) {
        String topic="";
        for (ConsumerRecord record: list) {
            topic = record.topic();
            log.info("handle real time message information：topic: {}, offset: {}, key: {}", topic, record.offset(), record.key());
            ReqStreamData reqStreamData = (ReqStreamData) record.value();
            if (null != reqStreamData) {
                RealTimeInfoReq realTimeInfoReq = new RealTimeInfoReq();
                try {
                    DataDecoder<RealTimeInfoReq> decoder = new DataDecoder<RealTimeInfoReq>();
                    decoder.decode(realTimeInfoReq, reqStreamData.getDataBytes());
                } catch (Exception e) {
                    log.error("issue real time message:---------->reqStreamData's dataBytes property raw value: {}",
                            ByteArrayUtil.toHexString(reqStreamData.getDataBytes()));
                    continue;
                }
                VehicleGpsReq vehicleGpsReq = realTimeInfoReq.vehicleGpsReq;
                DateTimeTag dateTimeTag = realTimeInfoReq.getCollectDate();
                if(dateTimeTag == null){
                    log.error("issue real time message:---------->reqStreamData's collect date is null");
                    continue;
                }
                double longitude = null == vehicleGpsReq ? 0.00 : vehicleGpsReq.longitude;
                if (longitude >= 73 && longitude <= 135) {
                    final String rowKey = getRowkey(record.key().toString(), dateTimeTag.getDate());
                    try {
                        ConcurrentMap<String, Object> batchMap = new ConcurrentHashMap<>(1);
                        Map<String, Object> messageContents = new HashMap<>(1);
                        messageContents.put("messageContent", reqStreamData.dataBytes);
                        batchMap.put(rowKey, messageContents);
                        disruptor.publish(rowKey.substring(0, rowKey.indexOf("_")), batchMap);
                    } catch (Exception e) {
                        log.error("handle pool found a exception, cause:", e);
                    }
                }else {
                    log.error("issue real time message:---------->reqStreamData's dataBytes property raw value: ",
                            ByteArrayUtil.toHexString(reqStreamData.getDataBytes()));
                }
            }else {
                log.error("[Vehicle topic: {}]----------> topic real time message is null", topic);
            }
        }
        acknowledgment.acknowledge();
    }

    /**
     * 部分报文相同复杂的处理逻辑
     * @param list
     * @param acknowledgment
     * @param hbaseTable
     * @param businessHbaseTable
     */
    public void complexCommonProcess(List<ConsumerRecord<String, Object>> list, Acknowledgment acknowledgment,String hbaseTable,String businessHbaseTable) {
        //存储容器 <rowkey,<colum,value>>
        HashMap<String, HashMap<String, Object>> topicMap = new HashMap<>(0);
        HashMap<String,HashMap<String,Object>> businessMap=new HashMap<>(0);
        String topic="";
        for (ConsumerRecord record:list) {
            log.info("处理实时报文：offset:{},key:{}",record.offset(),record.key());
            topic = record.topic();
            ReqStreamData reqStreamData = (ReqStreamData) record.value();
            if (null != reqStreamData) {
                String rowKey = getRowkey(record.key().toString(), reqStreamData.getCollectDate());
                HashMap<String, Object> map = new HashMap<String, Object>(1);
                map.put("messageContent", reqStreamData.dataBytes);
                //实时消息处理
                RealTimeInfoReq realTimeInfoReq = new RealTimeInfoReq();
                try {
                    DataDecoder<RealTimeInfoReq> decoder = new DataDecoder<RealTimeInfoReq>();
                    decoder.decode(realTimeInfoReq, reqStreamData.getDataBytes());
                } catch (Exception e) {
                    log.error("问题报文:---------->reqStreamData.getDataBytes={}", ByteArrayUtil.toHexString(reqStreamData.getDataBytes()));
                    continue;
                }
                //若报文的collectDate为空,则抛弃该报文，否则收集该报文
                VehicleGpsReq vehicleGpsReq = realTimeInfoReq.vehicleGpsReq;
                double longitude = null == vehicleGpsReq ? 0.00 : vehicleGpsReq.longitude;
                if (null != realTimeInfoReq.getCollectDate() && (longitude >= 73 && longitude <= 135)) {
                    //存储原始报文
                    topicMap.put(rowKey, map);

                    //存入业务数据
//                    HashMap<String, Object> businessMapRealTime = new HashMap<>(0);
//                    businessMapRealTime.put("messageContent", reqStreamData.dataBytes);
//                    RealTimeInfoReq realTimeReq = new RealTimeInfoReq();
//                    saveBusinessInfo(realTimeReq, reqStreamData, rowKey, businessMapRealTime, businessMap, record);

                } else {
                    log.error("问题报文-------------->{}", ByteArrayUtil.toHexString(reqStreamData.getDataBytes()));
                }
            }else {
                log.error("[车辆topic={}主题报文为空]", topic);
            }
        }

        //保存入hbase中
        try {
            if (!businessMap.isEmpty()) {
//                insertBatchData(businessHbaseTable, businessMap);
                acknowledgment.acknowledge();
                //log.info("{}---------------成功完成消费{}条数据",topic,list.size());
            }
        } catch (Exception e) {
            log.error("hbase保存{}业务数据共{}条失败.....{}",topic,list.size(), e);
        }
        if(!topicMap.isEmpty()){
            try {
                insertBatchData(hbaseTable,topicMap);
            } catch (Exception e) {
                log.error("hbase保存{}原始报文数据共{}条失败.....{}",topic,list.size(), e);
                e.printStackTrace();
            }
        }

    }

    private void saveBusinessInfo(RealTimeInfoReq realTimeReq,ReqStreamData reqStreamData,String rowKey,HashMap<String,Object>map,HashMap<String,HashMap<String,Object>> contentMap,ConsumerRecord record){
        try {
            //解析ReqStreamData中的报文
            DataDecoder<RealTimeInfoReq> decoder = new DataDecoder<RealTimeInfoReq>();
            decoder.decode(realTimeReq, reqStreamData.getDataBytes());
            if(null!=realTimeReq.commandTag&&realTimeReq.commandTag.equals(CommandTag.REALTIME_INFO_REPORT)){
                String realTimeJson = buildJson(reqStreamData.getCollectDate(), realTimeReq);
                map.clear();
                map.put("messageContent",realTimeJson);
                contentMap.put(rowKey,map);
            }
        } catch (Exception e) {
            log.error("[车辆实时报文][decode error][vin={}][data={}][err={}]",record.key(),reqStreamData.getDataBytes(),e.getMessage());
        }
    }

    /**
     * 将报文对象转换为业务json
     * */
    private String buildJson(Date collectDate, RealTimeInfoReq realTimeInfoReq){

        if(collectDate!=null && realTimeInfoReq!=null){
            Map<String,Object> map = new HashMap<String, Object>();

            //采集时间
            map.put(Contact.COLLECT_DATE, collectDate);
            //上报时间
            map.put(Contact.REPORT_DATE,realTimeInfoReq.collectDate==null?null:realTimeInfoReq.collectDate.getDate());
            //GPS数据
            VehicleGpsReq gpsReq = realTimeInfoReq.getVehicleGpsReq();
            if (gpsReq!=null) {
                //map.put(Contact.VEHICLE_GPS,GsonUtil.toJsonWithSerializeNulls(gpsReq));
                map.put(Contact.VEHICLE_GPS,JsonUtils.toJson(gpsReq));
            }
            //整车数据
            VehicleAllReq allReq = realTimeInfoReq.getVehicleAllReq();
            if(allReq!=null){
                //map.put(Contact.VEHICLE_ALL,GsonUtil.toJsonWithSerializeNulls(allReq));
                map.put(Contact.VEHICLE_ALL,JsonUtils.toJson(allReq));
                //				BeanUtils.copyProperties(gpsReq, realTimeInfoDto.getVehicleAll());
            }

            //扩展数据
            ExtendReq extendReq = realTimeInfoReq.getExtendReq();
            if(extendReq!=null){
                //扩展车辆状态数据
                ExtendVehicleState extendVehicleState = extendReq.getextendVehicleState();
                if(extendVehicleState!=null){
                    //扩展车辆状态数据
                    //map.put(Contact.EXTEND_VEHICLE_STATE,GsonUtil.toJsonWithSerializeNulls(extendVehicleState));
                    map.put(Contact.EXTEND_VEHICLE_STATE,JsonUtils.toJson(extendVehicleState));

                }
            };
            return JsonUtils.toJson(map);
            //return GsonUtil.toJsonWithSerializeNulls(map);
        }
        return null;
    }

    /**
     * Initializing consumer common server after of trigger this properties set batch thread pool for disruptor.
     * @throws Exception Exception.
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        disruptor.disruptor(realtimeHandler);
    }
}